package org.zhuzx.util;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 基础消费者队列，内部自动运行动态数量的消费线程，需要子类提供消费逻辑。<br>
 * 注意：因为并发执行的缘故，并不能按照队列中的顺序完成消费任务，只保证先领取不保证先完成。
 * @author	zhuzx	2017年8月29日
 */
public abstract class AbstractConsumerQueue<E> {
	
	protected static final int DEFAULT_QUEUE_SIZE = (HardwareVariables.CPU_THREAD_NUM + 1) * 2;

	private ThreadPoolExecutor executor;

	AbstractConsumerQueue() {}
	
	protected void initExecutor(BlockingQueue<Runnable> workQueue) {
		executor = new ThreadPoolExecutor(1, HardwareVariables.CPU_THREAD_NUM+1, 30,
				TimeUnit.SECONDS, workQueue, getRejectedExecutionHandler());
		executor.prestartAllCoreThreads();
	}

	protected RejectedExecutionHandler getRejectedExecutionHandler() {
		return new BlockingPusher();
	}
	
	static class BlockingPusher implements RejectedExecutionHandler {
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			try {
				if (!executor.isShutdown()) {
					executor.getQueue().put(r);
				}
			} catch (InterruptedException e) {
				Thread.currentThread().interrupt();
				throw new RejectedExecutionException(
						"Executor was interrupted while the task was waiting to put on work queue", e);
			}
		}
	}
	
	/**
	 * 将任务元素放入队列，返回成功与否的标志位。
	 */
	public boolean pushIntoQueue(final E element) {
		Runnable task = new Runnable() {
			public void run() {
				process(element);
			}
		};
		return submitTask(task);
	}
	
	protected boolean submitTask(Runnable task) {
		try {
			executor.execute(task);
			return true;
		} catch (RejectedExecutionException e) {
			return false;
		}
	}
	
	/**
	 * 消费逻辑，由子类实现。
	 * @param element 队列元素
	 */
	protected abstract void process(E element);
	
}